use super::{
    frame::{make_simple_frame, new_id, Frame, ACK, NOTIFY, NOTIFY_ACK, PING, PONG, REQ, RES},
    get_conn_map,
    reader::read_frame,
    service::{login::wait_login},
};
use crate::{
    error::{new_err, AppErr},
    server::device::{frame::make_frame, reader::READ_TIMEOUT},
    utils::current_timestamp,
};
use dashmap::DashMap;
use serde::{de::DeserializeOwned, Serialize};
use std::{
    net::SocketAddr,
    sync::{
        atomic::{AtomicU32, AtomicU8, Ordering},
        Arc,
    },
    time::Duration,
};
use tokio::{
    io::{split, AsyncWriteExt, BufReader, ReadHalf, WriteHalf},
    net::TcpStream,
    sync::{mpsc, oneshot, Semaphore},
    time::timeout,
};

#[derive(Debug)]
pub struct Conn {
    tx: mpsc::Sender<Box<[u8]>>,
    exit_sem: Semaphore,
    pub device_id: i64,
    pub addr: SocketAddr,
    pub connect_timestamp: u64,
    pub ping_count: AtomicU32,
    seq: AtomicU8,
    res_map: DashMap<u16, oneshot::Sender<Frame>>,
}

pub type SharedConn = Arc<Conn>;

pub async fn new_conn(addr: SocketAddr, stream: TcpStream) -> Result<(), AppErr> {
    let (reader, mut writer) = split(stream);
    let mut reader = BufReader::new(reader);
    let device = wait_login(&mut reader, &mut writer).await?;
    let (tx, rx) = mpsc::channel::<Box<[u8]>>(64);

    let conn = Arc::new(Conn {
        tx,
        exit_sem: Semaphore::new(0),
        device_id: device.id,
        addr,
        ping_count: AtomicU32::new(0),
        connect_timestamp: current_timestamp(),
        seq: AtomicU8::new(0),
        res_map: DashMap::new(),
    });

    tokio::spawn(write_task(conn.clone(), writer, rx));
    tokio::spawn(read_task(conn.clone(), reader));

    get_conn_map().insert(device.id, conn);

    Ok(())
}

impl Conn {
    pub async fn notify_data(&self, cmd: u8, body: &[u8]) -> Result<(), AppErr> {
        let seq = 0u8;

        self.tx
            .send(make_frame(NOTIFY, seq, cmd, body))
            .await
            .map_err(|_| new_err("notify send err"))?;

        Ok(())
    }

    pub async fn notify<REQ: Serialize>(&self, cmd: u8, req: &REQ) -> Result<(), AppErr> {
        self.notify_data(
            cmd,
            &serde_cbor::to_vec(req).map_err(|_| new_err("cbor serial err"))?,
        )
        .await?;
        Ok(())
    }

    pub async fn notify_ack_data(&self, cmd: u8, body: &[u8]) -> Result<(), AppErr> {
        let seq = self.seq.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        self.res_map.insert(new_id(ACK, seq), tx);

        self.tx
            .send(make_frame(NOTIFY_ACK, seq, cmd, body))
            .await
            .map_err(|_| new_err("notify_ack send err"))?;

        _ = timeout(Duration::from_secs(3), rx)
            .await
            .map_err(|_| new_err("notify_ack req timeout"))?
            .map_err(|_| new_err("notify_ack rx err"))?;
        Ok(())
    }

    pub async fn notify_ack<REQ: Serialize>(&self, cmd: u8, req: &REQ) -> Result<(), AppErr> {
        self.notify_ack_data(
            cmd,
            &serde_cbor::to_vec(req).map_err(|_| new_err("cbor serial err"))?,
        )
        .await?;
        Ok(())
    }

    pub async fn req_data(&self, cmd: u8, body: &[u8]) -> Result<Frame, AppErr> {
        let seq = self.seq.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        self.res_map.insert(new_id(RES, seq), tx);

        self.tx
            .send(make_frame(REQ, seq, cmd, body))
            .await
            .map_err(|_| new_err("req send err"))?;

        let frame = timeout(Duration::from_secs(10), rx)
            .await
            .map_err(|_| new_err("simple req timeout"))?
            .map_err(|_| new_err("simple rx err"))?;

        Ok(frame)
    }

    pub async fn req<REQ: Serialize, RES: DeserializeOwned>(
        &self,
        cmd: u8,
        req: &REQ,
    ) -> Result<RES, AppErr> {
        let frame = self
            .req_data(
                cmd,
                &serde_cbor::to_vec(req).map_err(|_| new_err("cbor serial err"))?,
            )
            .await?;

        Ok(frame.parse()?)
    }

    pub async fn ping(&self) -> Result<(), AppErr> {
        let seq = self.seq.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        self.res_map.insert(new_id(PONG, seq), tx);

        self.tx
            .send(make_simple_frame(PING, seq))
            .await
            .map_err(|_| new_err("ping send err"))?;

        _ = timeout(Duration::from_secs(2), rx)
            .await
            .map_err(|_| new_err("ping rx timeout"))?
            .map_err(|_| new_err("ping rx err"))?;
        Ok(())
    }
}

async fn read_task(conn: SharedConn, mut r: BufReader<ReadHalf<TcpStream>>) {
    let ret = read_loop(&conn, &mut r).await;
    if let Err(e) = ret {
        println!("read err:{}", e);
    }
    _ = conn.exit_sem.add_permits(1);

    println!("read exit");
}

async fn read_loop(
    conn: &SharedConn,
    r: &mut BufReader<ReadHalf<TcpStream>>,
) -> Result<(), AppErr> {
    loop {
        let frame = tokio::select! {
            _ = conn.exit_sem.acquire() => { break; }
            frame = timeout(READ_TIMEOUT, read_frame(r)) => { frame.map_err(|_| new_err("read timeout"))?? }
        };
        let c_type = frame.c_type;

        match c_type {
            PING => {
                conn.ping_count.fetch_add(1, Ordering::SeqCst);
                conn.tx
                    .try_send(make_simple_frame(PONG, frame.seq))
                    .map_err(|_| new_err("send pong err"))?;
            }

            PONG | RES | ACK => {
                let tx = conn.res_map.remove(&frame.id());
                if let Some(tx) = tx {
                    _ = tx.1.send(frame);
                }
            }

            NOTIFY_ACK => {
                conn.tx
                    .try_send(make_simple_frame(ACK, frame.seq))
                    .map_err(|_| new_err("send ack err"))?;
            }

            _ => return Err(new_err("unknown type")),
        };
    }

    Ok(())
}

async fn write_loop(
    conn: &SharedConn,
    w: &mut WriteHalf<TcpStream>,
    r: &mut mpsc::Receiver<Box<[u8]>>,
) -> Result<(), AppErr> {
    loop {
        let frame = tokio::select! {
            _ = conn.exit_sem.acquire() => { break; }
            frame = r.recv() => { frame }
        };
        let frame = frame.ok_or(new_err("recv none"))?;
        w.write_all(&frame).await?;
    }

    Ok(())
}

async fn write_task(
    conn: SharedConn,
    mut writer: WriteHalf<TcpStream>,
    mut rx: mpsc::Receiver<Box<[u8]>>,
) {
    let ret = write_loop(&conn, &mut writer, &mut rx).await;
    if let Err(e) = ret {
        println!("write err:{}", e);
    }
    _ = conn.exit_sem.add_permits(1);
    get_conn_map().remove(&conn.device_id);
    println!("write exit");
}
